package com.example.test.dialogfactory;

import android.content.Context;
import android.content.Intent;
import android.os.Bundle;
import androidx.appcompat.app.AppCompatActivity;
import easier.log.logger.Logger;
import io.reactivex.rxjava3.annotations.NonNull;
import io.reactivex.rxjava3.core.Observable;
import io.reactivex.rxjava3.core.ObservableSource;
import io.reactivex.rxjava3.core.Observer;
import io.reactivex.rxjava3.disposables.Disposable;
import io.reactivex.rxjava3.functions.Consumer;
import io.reactivex.rxjava3.functions.Function;
import io.reactivex.rxjava3.subjects.PublishSubject;
import java.util.concurrent.TimeUnit;

public class MainActivity30 extends AppCompatActivity {

    public static void start( Context context ) {

        Intent starter = new Intent( context, MainActivity30.class );
        context.startActivity( starter );
    }

    @Override
    protected void onCreate( Bundle savedInstanceState ) {

        super.onCreate( savedInstanceState );
        setContentView( R.layout.activity_main30 );

        ObservableContainer< String > container = new ObservableContainer<>();

        container.getObservable()
            .flatMap( new Function< String, ObservableSource< ? > >() {
                @Override
                public ObservableSource< ? > apply( String s ) throws Throwable {

                    PublishSubject< String > subject = PublishSubject.create();
                    Logger.d( "flat:" + s );

                    new Thread( new Runnable() {
                        @Override
                        public void run() {

                            try {
                                int i = Integer.parseInt( s ) * 1000;
                                Thread.sleep( i );
                            } catch ( Exception e ) {
                                e.printStackTrace();
                            }

                            subject.onNext( "new " + s );
                            subject.onComplete();
                        }
                    } ).start();

                    return subject;
                }
            } )
            .subscribe( new Observer< Object >() {
                @Override
                public void onSubscribe( @NonNull Disposable d ) {

                }

                @Override
                public void onNext( @NonNull Object o ) {

                    Logger.d( "result:" + o );
                }

                @Override
                public void onError( @NonNull Throwable e ) {

                    Logger.d( "error:" + e.getMessage() );
                }

                @Override
                public void onComplete() {

                    Logger.d( "all finished" );
                }
            } );

        Observable.timer( 5, TimeUnit.SECONDS )
            .subscribe( new Consumer< Long >() {
                @Override
                public void accept( Long aLong ) throws Throwable {

                    container.getSubscriber().onNext( "2" );
                    container.getSubscriber().onNext( "4" );
                    container.getSubscriber().onNext( "6" );
                    container.getSubscriber().onComplete();

                    Logger.d( "finished" );
                }
            } );
    }

    public static class ObservableContainer< E > {

        private final PublishSubject< E > mSubscriber = PublishSubject.create();

        private final Observable< E > mObservable = mSubscriber;

        public Observable< E > getObservable() {

            return mObservable;
        }

        public Observer< E > getSubscriber() {

            return mSubscriber;
        }
    }
}